package com.sunsys.mapreduce.examples.counters;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.CounterGroup;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

public class AverageMapperDriver extends Configured implements Tool {

	@Override
	public int run(String[] arg0) throws Exception {
		String inputFile = "/home/ubuntu/workspace-info/CountersExample/input/InputFile.txt";
		String outputPath = "/home/ubuntu/workspace-info/CountersExample/output/1";
		
		
		Configuration conf=new Configuration();
		Job job = new Job(conf, "AverageCounterJob");
		
		job.setJarByClass(AverageMapperDriver.class);
		job.setMapperClass(AverageMapper.class);
//		job.setReducerClass(cls);		//default is Identity Reducer
//		job.setNumReduceTasks(0);	//Indicates no need to Run the Reducer phase
		
		FileInputFormat.addInputPath(job, new Path(inputFile));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));

		job.setMapOutputKeyClass(NullWritable.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		boolean result = job.waitForCompletion(true);
		
		Counters counters = job.getCounters();
		/* Retrieve the counter built-in counter value */
		long inputRecordsCount = counters.findCounter("org.apache.hadoop.mapred.Task$Counter", "MAP_INPUT_RECORDS").getValue();
		/* Retrieve the user defined dynamic counter value */
		long counter1Value = counters.findCounter("com.sunsys.examples.dynamic.countergroup", "counter1").getValue();

		CounterGroup counterGroup = null;
		
		System.out.println("No.of input records:"+inputRecordsCount);
		System.out.println("Counter1 value:"+counter1Value);

		return 0;
	}
	
	public static void main(String []args) throws IOException, InterruptedException, ClassNotFoundException, Exception {
		AverageMapperDriver driver = new AverageMapperDriver();
		driver.run(args);
	}

}
